package com.shujia.stream

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object CardAvgSpeedFlow {
  def main(args: Array[String]): Unit = {


    //环境设置对象
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()

    /**
     * flink sql环境
     *
     */
    val table: TableEnvironment = TableEnvironment.create(settings)


    /**
     * 创建source表，增加事件时间字段
     *
     */
    table.executeSql(
      """
        |CREATE TABLE cars (
        |    car STRING,
        |    city_code STRING,
        |    county_code STRING,
        |    card BIGINT,
        |    camera_id STRING,
        |    orientation STRING,
        |    road_id BIGINT,
        |    `time` BIGINT,
        |    speed DOUBLE,
        |    ts as TO_TIMESTAMP(FROM_UNIXTIME(`time`)),
        |    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
        |) WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'car',
        |  'properties.bootstrap.servers' = 'master:9092',
        |  'properties.group.id' = 'testGroup',
        |  'scan.startup.mode' = 'earliest-offset',
        |  'format' = 'json'
        |)
        |
        |""".stripMargin)

    table.executeSql(
      """
        |CREATE TABLE card_avg_speed_and_flow (
        |    `card` BIGINT,
        |    `window_start` TIMESTAMP(3) ,
        |    `window_end` TIMESTAMP(3) ,
        |    avg_speed DOUBLE,
        |    flow BIGINT,
        |    PRIMARY KEY (card,`window_start`,`window_end`) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/car',
        |   'table-name' = 'card_avg_speed_and_flow',
        |    'username' ='root',
        |    'password'='123456'
        |)
        |""".stripMargin)


    table.executeSql(
      """
        |
        |insert into card_avg_speed_and_flow
        |select
        |card,
        |HOP_START(ts, INTERVAL  '5' MINUTES, INTERVAL  '15'  MINUTES) as window_start,
        |HOP_end(ts, INTERVAL  '5' MINUTES, INTERVAL  '15'  MINUTES) as window_end,
        |round(avg(speed) , 2)  as avg_speed,
        |count(1) as flow
        |from
        |cars
        |group by
        |card,
        |HOP(ts, INTERVAL  '5' MINUTES, INTERVAL  '15'  MINUTES)
        |
        |
        |""".stripMargin)


  }

}
